/**
 * Created by hadoop on 21-5-15.
 */
import java.util.Arrays;
import java.util.Iterator;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import scala.Tuple2;

public class wordCountJava {
    //main函数
    public static void main(String[] args) {
        //创建SparkConf对象，设置Spark应用的配置信息
        SparkConf conf = new SparkConf()
                .setAppName("wordCountJava")
                .setMaster("local");//本地运行

        //创建JavaSparkContext对象
        JavaSparkContext sc = new JavaSparkContext(conf);

        //Java中创建的普通RDD叫做JavaRDD
        JavaRDD<String> lines = sc.textFile("hdfs://localhost:9000/user/hadoop/localReceive/week2/week2Input.txt");

        //传递给flatMap函数一个匿名内部类的实例，将每一行拆分成单个的单词
        JavaRDD<String> words = lines.flatMap(new FlatMapFunction<String, String>() {
            public Iterator<String> call(String line) throws Exception {
                return Arrays.asList(line.split(" ")).iterator();
            }
        });
        //传递给mapToPair一个匿名内部类的实例，将每一个单词映射为(单词, 1)格式
        JavaPairRDD<String, Integer> pairs = words.mapToPair(
                new PairFunction<String, String, Integer>() {
                    public Tuple2<String, Integer> call(String word) throws Exception {
                        return new Tuple2<String, Integer>(word, 1);
                    }
                });

        //传递给reduceByKey一个匿名内部类的实例，将相同key的两个值进行相加
        JavaPairRDD<String, Integer> wordCounts = pairs.reduceByKey(
                new Function2<Integer, Integer, Integer>() {
                    public Integer call(Integer v1, Integer v2) throws Exception {
                        return v1 + v2;
                    }
                });

        //传递给foreach函数一个匿名内部类的实例，该实例主要负责打印输出
        wordCounts.foreach(new VoidFunction<Tuple2<String,Integer>>() {
            public void call(Tuple2<String, Integer> wordCount) throws Exception {
                //Tuple2是Scala中的元组
                System.out.println(wordCount._1 + "  " + wordCount._2);
            }
        });
        sc.close();
    }
}

